package cn.ac.iie.di.ban.ycl.cleaner.handler;

import cn.ac.iie.di.ban.ycl.cleaner.common.Constants;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import org.apache.commons.configuration2.Configuration;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.http.client.fluent.Request;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author 张明诚
 * @date 2016-11-18 10:39:47
 * @version V1.0
 * @desc
 */
public class DeduplicationFilter implements FilterInterface {

	private final Logger LOG = LoggerFactory.getLogger(DeduplicationFilter.class);
	private String[] columnNames;
	private REFieldType[] types;
	private String docSchemaName;
	private Map<String, String> userDesc;
	private List<Object[]> datas;
	private final String[] urls;
	private final List<Object> deduplicationFields;
	private final String prefix;

	public DeduplicationFilter(Configuration configuration) {
		this.urls = configuration.getStringArray(Constants.DEDUPLICATION_URL);
		this.deduplicationFields = configuration.getList(Constants.DEDUPLICATION_FIELDS);
		prefix = configuration.getString(Constants.DEDUPLICATION_PREFIX, "");
	}

	@Override
	public REMessage filter(REMessage message) throws Exception {
		columnNames = message.colNames;
		types = message.types;
		docSchemaName = message.docSchemaName;
		userDesc = message.userDesc;
		datas = message.data;
		String deduplicationValues = getDeduplicationValues();
		if (deduplicationValues.isEmpty()) {
			throw new Exception("deduplication values is empty!");
		} else {
			LOG.debug("deduplication values: \n{}", deduplicationValues);
		}
		String response = deduplicationService(deduplicationValues);
		if (response.isEmpty()) {
			throw new Exception("deduplication response values is empty!");
		} else {
			LOG.debug("deduplication result:\n{}", response);
		}
		String[] split = response.split("\n");
		List<String> flags = Arrays.asList(split);
		deleteDuplicationDatas(flags);
		REMessage deduplicationMessage
				= new REMessage(columnNames, types, datas, docSchemaName, userDesc);
		return deduplicationMessage;
	}

	public String getDeduplicationValues() {
		StringBuilder sb = new StringBuilder();
		List<String> columnList = Arrays.asList(columnNames);
		for (Object[] data : this.datas) {
			boolean isFirst = true;
			for (Iterator<Object> it = deduplicationFields.iterator(); it.hasNext();) {
				String distinctField = (String) it.next();
				int index = columnList.indexOf(distinctField);
				if (index < 0) {
					LOG.info(distinctField);
					LOG.error("the distinct field: {} does not exists!", distinctField);
				}
				Object distinctFieldValue = data[index];
				if (distinctFieldValue == null || distinctFieldValue.toString().isEmpty()) {
					LOG.info("columnames:{}", Arrays.toString(columnNames));
					LOG.info("data:{}", Arrays.toString(data));
					LOG.error("the value of distinct field: {} is null or empty!", distinctField);
				}
				if (!isFirst) {
					sb.append("^");
				}
				sb.append(prefix).append(distinctFieldValue);
				isFirst = false;
			}
			sb.append("\n");
		}
		LOG.debug("deduplicate string: " + sb.toString());
		return sb.toString();
	}

	public void deleteDuplicationDatas(List<String> flags) {
		List<Object[]> newMessage = new ArrayList<>();
		for (int i = 0; i < flags.size(); i++) {
			if (flags.get(i).equals("0")) {
				newMessage.add(datas.get(i));
			}
		}
		this.datas = newMessage;
	}

	private String deduplicationService(String content) {
//		if (true) {
//			int n = content.trim().split("\n").length;
//
//			if (n > 0) {
//				StringBuilder sb = new StringBuilder();
//				sb.append('0');
//				for (int i = 1; i < n; ++i) {
//					sb.append("\n0");
//				}
//				return sb.toString();
//			} else {
//				return "";
//			}
//		}
		String response = null;
		String url;
		int index;
		while (true) {
			index = new Random().nextInt(urls.length);
			url = urls[index];
			try {
				response = Request.Post(url)
						.bodyByteArray(content.getBytes())
						//.connectTimeout(3000)
						.connectTimeout(1800000)//30 min
						//.socketTimeout(5000)
						.socketTimeout(1800000)//30 min
						.execute()
						.returnContent().asString();
			} catch (IOException ex) {
				LOG.warn("request url {} error! \n"
						+ ExceptionUtils.getFullStackTrace(ex), url);
			}
			if (response == null || response.trim().isEmpty()) {
				LOG.error("response content is null，please check the url:{}", url);
			} else {
				break;
			}
			try {
				Thread.sleep(5000);
			} catch (InterruptedException ex) {
			}
		}
		return response;
	}

	private String deduplicationServiceNew(String content) {
		//list of ids remain to be searched
		List<String> remainSearchList = Arrays.asList(content.split("\n"));
		//map from id to its index
		HashMap<String, Integer> idxMap = new HashMap<>();
		for (int i = 0; i < remainSearchList.size(); ++i) {
			idxMap.put(remainSearchList.get(i), i);
		}
		//result list for search results which will be translated to string.
		List<String> resultList = new ArrayList<>(idxMap.size());
		//search respponse
		String response = null;
		String url;
		int index;
		while (!remainSearchList.isEmpty()) {
			//do one search
			while (true) {
				//choose a url
				index = new Random().nextInt(urls.length);
				url = urls[index];
				//build a content
				StringBuilder sb = new StringBuilder();
				for (String s : remainSearchList) {
					sb.append(s).append("\n");
				}
				content = sb.toString();
				//search
				try {
					response = Request.Post(url)
							.bodyByteArray(content.getBytes())
							.connectTimeout(3000)
							.socketTimeout(5000)
							.execute()
							.returnContent().asString();
				} catch (IOException ex) {
					LOG.warn("request url {} error! \n"
							+ ExceptionUtils.getFullStackTrace(ex), url);
				}
				if (response == null || response.trim().isEmpty()) {
					LOG.error("response content is null，please check the url:{}", url);
				} else {
					break;
				}
				try {
					Thread.sleep(5000);
				} catch (InterruptedException ex) {
				}
			}//end while true for total-fail search
		}//end while remainSearchList not empty for partly-fail search
		return response;
	}

}
